#include "Spark/Net/TcpServer.h"
#include "Spark/Net/Impl/Acceptor.h"
#include "Spark/Utils/Logger.h"
#include "Spark/Net/EventLoop.h"
#include "Spark/Net/EventLoopThreadPool.h"

TcpServer::TcpServer(EventLoop* loop, const SocketAddr& addr, std::string name)
    : loop_((assert(loop), loop)),
      ipPort_(std::move(addr.toStringPort())),
      serverName_(std::move(name)),
      acceptorPtr_(new Acceptor(loop, addr)),
      messageCallback_(
          [](const TcpConnectionPtr&, MsgBuffer* buffer, TimePoint tp) {
              LOG_ERROR << "default handle recv message ["
                        << buffer->readableBytes() << " bytes]";
              buffer->retrieveAll();
          })
{
    acceptorPtr_->setNewConnectionCallback(
        [this](int sockfd, const SocketAddr& sa) {
            newConnection(sockfd, sa);
        });
}

TcpServer::~TcpServer()
{
    LOG_TRACE << "TcpServer::~TcpServer [" << serverName_ << "]";
}

void TcpServer::start()
{
    if (!started_)
    {
        threadPool_->start();
        assert(!acceptorPtr_->listening());
        auto accptr = acceptorPtr_.get();
        loop_->runInLoop([accptr] { accptr->listen(); });
    }
}

void TcpServer::setIoLoopNum(size_t num)
{
    assert(!started_);
    static bool is_set = false;
    if (!is_set)
    {
        threadPool_ =
            std::make_shared<EventLoopThreadPool>(loop_, "IoLoopPool");
        is_set = true;
    }
}

void TcpServer::newConnection(int sockfd, const SocketAddr& peer)
{
    assert(loop_->isInLoopThread());
    EventLoop* loop = threadPool_->getNextLoop();
    char connName[128]{};
    snprintf(connName,
             sizeof(connName) - 1,
             "%s-%s#%d",
             serverName_.c_str(),
             ipPort_.c_str(),
             nextConnId_);
    ++nextConnId_;
    LOG_INFO << "TcpServer::newConnection [" << serverName_
             << "] accept new connection [" << connName << "] from "
             << peer.toStringPort();
    SocketAddr local = SocketAddr::getLocalAddr(sockfd);
    auto conn =
        std::make_shared<TcpConnection>(loop, sockfd, connName, local, peer);
    connections_[connName] = conn;
    conn->setConnectionCallback(connectionCallback_);
    conn->setMessageCallback(messageCallback_);
    conn->setWriteDoneCallback(writeDoneCallback_);
    conn->setCloseCallback(
        [this](const TcpConnectionPtr& conn) { removeConnection(conn); });
    loop->runInLoop([conn] { conn->connectEstablished(); });
}

void TcpServer::removeConnection(const TcpConnectionPtr& conn)
{
    loop_->runInLoop([this, conn] { removeConnectionInLoop(conn); });
}

void TcpServer::removeConnectionInLoop(const TcpConnectionPtr& conn)
{
    assert(loop_->isInLoopThread());
    size_t n = connections_.erase(conn->name());
    LOG_INFO << "TcpServer::removeConnectionInLoop remove[" << n
             << "] connections";
    assert(n == 1);
    EventLoop* loop = conn->getLoop();
    loop->queueInLoop([conn] { conn->connectDestroyed(); });
}